#include <pica_jsonrpc.hh>
#include <pica_statdb.hh>
#include "config.h"

#ifdef __cplusplus
extern "C" {
#endif
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <ctype_pica_lcmgr_api.h>
#include <ev.h>
#include <jsonrpc.h>
#include <api/yajl_common.h>
#include <api/yajl_tree.h>
#ifdef __cplusplus
}
#endif

#define JSONRPCSVRPRT 8332
#define JSONRPCBUFSZ 1500
#define RFCTIMESTRLEN 40
#define MAX_JSON_OBJ_LEN 4*JSONRPCBUFSZ
#define CONNUMLIMITS 5

// enhance LATER
typedef struct {
    int interval;
    int repeat;
    time_t when_utc;
    bool write_to_file;
} Schedule;

typedef struct resp_data {
    void (* print_result)(void *ctx, const char *fmt, ...);
    void *ctx;
    int rows;
} RespData;

struct ev_loop *loop;
struct jrpc_server pica_jrpc_server;
jsonrpc_server_t *jsonrpc_server;

Schedule schedule;
ev_timer timeout_watcher;

bool
pica_is_numeric(const char *p)
{
    char c;
    if (p == NULL) return false;
    while (c=*p++) {
        if (!isdigit(c)) return false;
    }
    return true;
}

static void
pica_jrpc_timeout_cb(EV_P_ struct ev_timer* timer, int revents)
{
    schedule.repeat--;
    if (schedule.repeat <= 0) 
        ev_timer_stop(loop, &timeout_watcher);
    pica_stat_sync_requested = true;
    pica_stat_write_file_flag = schedule.write_to_file;
    ctype_pica_stat_sync(schedule.interval); // ignore error
    pica_stat_write_file_flag = false;
    pica_stat_sync_requested = false;
    return;
}

int
pica_jrpc_get_counters_cb(void *p_data, int num_fields, char **p_fields, char **p_col_names)
{
    int i;
    RespData *resp = (RespData *)p_data;

    if (num_fields <= 0) {
        resp->print_result(resp->ctx, "[]");
        resp->rows++;
        return 0;
    }
    if (resp->rows == 0) {
        resp->print_result(resp->ctx, "[\"%s\"", p_col_names[0]);
        for (i = 1; i < num_fields; i++) {
            resp->print_result(resp->ctx, ",\"%s\"", p_col_names[i]);
        }
        resp->print_result(resp->ctx, "]");
        resp->rows++;
    }
    if (pica_is_numeric(p_fields[0]))
        resp->print_result(resp->ctx, ",[%s", p_fields[0]);
    else
        resp->print_result(resp->ctx, ",[\"%s\"", p_fields[0]);
    for (i = 1; i < num_fields; i++) {
        if (pica_is_numeric(p_fields[i]))
            resp->print_result(resp->ctx, ",%s", p_fields[i]);
        else
            if ((p_fields[i] != NULL) &&
                    ('[' == *p_fields[i] || '{' == *p_fields[i]))
                resp->print_result(resp->ctx, ",%s", p_fields[i]);
            else
                resp->print_result(resp->ctx, ",\"%s\"", p_fields[i]);

    }
    resp->print_result(resp->ctx, "]");
    resp->rows++;
    return 0;
}

jsonrpc_error_t
pica_jrpc_get_counters(int argc, const jsonrpc_param_t *argv,
                       void (* print_result)(void *ctx, const char *fmt, ...),
                       void *ctx)
{
    bool rc;
    int  ret;
    char *errmsg = 0;
    RespData resp;

    resp.print_result = print_result;
    resp.ctx = ctx;
    resp.rows = 0;
    print_result(ctx, "[");
    rc = pica_db_execute_sql(argv->json.u.string, pica_jrpc_get_counters_cb,
                             &resp, &errmsg);
    if (errmsg) {
        print_result(ctx, "\"%s\"", errmsg);
        pica_db_free((void *)errmsg);
    }
    print_result(ctx, "]");
    return JSONRPC_ERROR_OK;
}

jsonrpc_error_t
pica_jrpc_prepare_counters(int argc, const jsonrpc_param_t *argv,
                           void (* print_result)(void *ctx, const char *fmt, ...),
                           void *ctx)
{
    char timestamp_str[40];
    char rfcTime[RFCTIMESTRLEN];
    struct tm tm, *t;
    int i;
    ev_tstamp after, now, delay;
    bool rc;

    bzero(&tm, sizeof(tm));
    for (i = 0; i < argc; i++) {
        if (strcmp(argv[i].name, "when") == 0) {
            strptime(argv[i].json.u.string, "%Y-%m-%dT%H:%M:%S", &tm);
            schedule.when_utc = mktime(&tm);
            if (schedule.when_utc < 0) {
                return JSONRPC_ERROR_PARAM_OUT_OF_RANGE;
            }
        } else if (strcmp(argv[i].name, "interval") == 0) {
            schedule.interval = argv[i].json.u.number;
            if (schedule.interval < 0) {
                return JSONRPC_ERROR_PARAM_OUT_OF_RANGE;
            }
        } else if (strcmp(argv[i].name, "repeat") == 0) {
            schedule.repeat = argv[i].json.u.number;
            if (schedule.repeat < 0) {
                return JSONRPC_ERROR_PARAM_OUT_OF_RANGE;
            }
        } else if (strcmp(argv[i].name, "write-to-file") == 0) {
            schedule.write_to_file = argv[i].json.u.boolean;
        } else if (strcmp(argv[i].name, "sql") == 0) {
            //LATER
        }
    }
    
    /* if 'when' has passed, take a sample now and return
       the current timestamp in 'result'.
       if 'when' is yet to come, take a sample at 'when' and
       return the 'when' in 'result'.
       Both ev_tstamp and time_t represent secs after 01/01/1970.
     */
    now = ev_now(loop);
    after = (ev_tstamp)schedule.when_utc - now;
    if (after <= 0) {
        /* in Broadcom case, this will trigger soc_counter_thread to
           wake up and read h/w counters and trigger our callback, which
           may create tables in database and update 'lastTimestamp'.
           ctype_pica_stat_sync() does not return until soc_counter_thread
           has triggered our callback.
         */
        if (schedule.repeat > 0) {
            /* wait until the start of a second */
            ev_sleep(delay = (1.0 + (int)now) - now);
        }

        pica_stat_sync_requested = true;
        pica_stat_write_file_flag = schedule.write_to_file;
        rc = ctype_pica_stat_sync(schedule.interval);
        pica_stat_write_file_flag = false;
        pica_stat_sync_requested = false;
        if (rc == false) {
            return JSONRPC_ERROR_SERVER_INTERNAL;
        }

        t = gmtime(&lastStartTimestamp.tv_sec);
        snprintf(rfcTime, RFCTIMESTRLEN, "%04d-%02d-%02dT%02d:%02d:%02dZ",
                 t->tm_year+1900,  t->tm_mon+1, t->tm_mday, t->tm_hour,
                 t->tm_min, t->tm_sec);
        print_result(ctx, "\"%s\"", rfcTime);

        if (schedule.repeat <= 0) {
            ev_timer_stop(loop, &timeout_watcher);
            return JSONRPC_ERROR_OK;
        }

        ev_timer_init(&timeout_watcher, pica_jrpc_timeout_cb,
                      (ev_tstamp)schedule.interval + delay,
                      (ev_tstamp)schedule.interval);
        ev_timer_start(loop, &timeout_watcher);
    } else {
        ev_timer_init(&timeout_watcher, pica_jrpc_timeout_cb,
                      after, (ev_tstamp)schedule.interval);
        ev_timer_start(loop, &timeout_watcher);

        t = gmtime(&schedule.when_utc);
        snprintf(rfcTime, RFCTIMESTRLEN, "%04d-%02d-%02dT%02d:%02d:%02dZ",
                 t->tm_year+1900,  t->tm_mon+1, t->tm_mday, t->tm_hour,
                 t->tm_min, t->tm_sec);
        print_result(ctx, "\"%s\"", rfcTime);
    }
    return JSONRPC_ERROR_OK;
}

static jsonrpc_handle_t
jsonrpc_yajl_parse(const char *json)
{
    yajl_val root = yajl_tree_parse(json, NULL, 0);
    return (jsonrpc_handle_t)root;
}

static void
jsonrpc_yajl_release(jsonrpc_handle_t json)
{
    yajl_tree_free((yajl_val)json);
}

static jsonrpc_handle_t
jsonrpc_yajl_get(jsonrpc_handle_t json, const char *key)
{
    const char *path[2];

    path[0] = key;
    path[1] = NULL;
    return (jsonrpc_handle_t)yajl_tree_get((yajl_val)json, path, yajl_t_any);
}

static jsonrpc_handle_t
jsonrpc_yajl_get_at(jsonrpc_handle_t json, size_t index)
{
    yajl_val v = (yajl_val)json;

    if (v->type == yajl_t_object) {
        if (v->u.object.len > index)
            return (jsonrpc_handle_t)v->u.object.values[index];
    } else if (v->type == yajl_t_array) {
        if (v->u.array.len > index)
            return (jsonrpc_handle_t)v->u.array.values[index];
    }
    return NULL;
}

static const char *
jsonrpc_yajl_get_key_at(jsonrpc_handle_t json, size_t index)
{
    yajl_val v = (yajl_val)json;

    if (v->type == yajl_t_object) {
        if (v->u.object.len > index)
            return v->u.object.keys[index];
    }
    return NULL;
}

static jsonrpc_bool_t
jsonrpc_yajl_valueof (jsonrpc_handle_t json, jsonrpc_json_t *value)
{
    yajl_val v = (yajl_val)json;

    switch (v->type) {
    case yajl_t_string:
        value->type = JSONRPC_TYPE_STRING;
        if (v->u.string == NULL)
            return JSONRPC_FALSE;
        value->u.string = v->u.string;
        break;

    case yajl_t_number:
        value->type = JSONRPC_TYPE_NUMBER;
        if (v->u.number.flags & YAJL_NUMBER_DOUBLE_VALID)
            value->u.number = v->u.number.d;
        else if (v->u.number.flags & YAJL_NUMBER_INT_VALID)
            value->u.number = (double)v->u.number.i;
        else
            return JSONRPC_FALSE;
        break;

    case yajl_t_object:
        value->type = JSONRPC_TYPE_OBJECT;
        value->u.object = (jsonrpc_handle_t)json;
        break;

    case yajl_t_array:
        value->type = JSONRPC_TYPE_ARRAY;
        value->u.array = (jsonrpc_handle_t)json;
        break;

    case yajl_t_true:
        value->type = JSONRPC_TYPE_BOOLEAN;
        value->u.boolean = JSONRPC_TRUE;
        break;

    case yajl_t_false:
        value->type = JSONRPC_TYPE_BOOLEAN;
        value->u.boolean = JSONRPC_FALSE;
        break;
    case yajl_t_null:
        value->type = JSONRPC_TYPE_NULL;
        break;

    default:
        return JSONRPC_FALSE;
    }
    return JSONRPC_TRUE;
}

static size_t
jsonrpc_yajl_length(jsonrpc_handle_t json)
{
    yajl_val v = (yajl_val)json;

    if (v->type == yajl_t_object) {
        return v->u.object.len;
    } else if (v->type == yajl_t_array) {
        return v->u.array.len;
    }
    return 0;
}

const jsonrpc_json_plugin_t*
jsonrpc_plugin_yajl (void)
{
    static const jsonrpc_json_plugin_t plugin_yajl = {
        jsonrpc_yajl_parse,
        jsonrpc_yajl_release,
        jsonrpc_yajl_get,
        jsonrpc_yajl_get_at,
        jsonrpc_yajl_get_key_at,
        jsonrpc_yajl_valueof,
        jsonrpc_yajl_length
    };
    return &plugin_yajl;
}


// get sockaddr, IPv4 or IPv6:
static void *
pica_get_in_addr(struct sockaddr *sa)
{
    if (sa->sa_family == AF_INET) {
        return &(((struct sockaddr_in*) sa)->sin_addr);
    }
    return &(((struct sockaddr_in6*) sa)->sin6_addr);
}

static int
pica_send_response(struct jrpc_connection * conn, const char *response)
{
    int fd = conn->fd;
    if (response == NULL)
        return 0; /* Do not send any response */
    if (conn->debug_level > 1)
        printf("JSON Response:\n%s\n", response);
    write(fd, response, strlen(response));
    write(fd, "\n", 1);
    return 0;
}

static void
pica_close_connection(struct ev_loop *loop, ev_io *w)
{
    ev_io_stop(loop, w);
    close(((struct jrpc_connection *) w)->fd);
    ((struct jrpc_server *) w->data)->con_number--;
    free(((struct jrpc_connection *) w)->buffer);
    free(((struct jrpc_connection *) w));
    if (((struct jrpc_server *) w->data)->con_number < CONNUMLIMITS &&
            !ev_is_active(&((struct jrpc_server *) w->data)->listen_watcher)) {
        ev_io_start(((struct jrpc_server *) w->data)->loop,
                &((struct jrpc_server *) w->data)->listen_watcher);
        printf("ev_io_loop started \n");
    }
}

static int
pica_parse_stream(char *value, const char *limit, char **end_ptr)
{
    int count = 0;

    /* search up to matching '[' and ']' pair or '{' and '}' pair */
    /* return -1 for errors, 0 for being unmatched, 1 for being matched */
    if (!end_ptr)
        return -1;
    if (!value || *value == 0)
        return -1;
    while ((*value <= ' ') && (value < limit)) value++;
    if (value >= limit)
        return -1;
    if (*value == '[') {
        while (value < limit) {
            value++;
            if (*value == ']') {
                if (count == 0) {
                    /* set end_ptr to end of matching ']' */
                    *end_ptr = value + 1;
                    return 1;
                }
                count--;
            } else if (*value == '[') {
                count++;
            }
        }
        return 0;
    }
    if (*value == '{') {
        while (value < limit) {
            value++;
            if (*value == '}') {
                if (count == 0) {
                    /* set end_ptr to end of matching '}' */
                    *end_ptr = value + 1;
                    return 1;
                }
                count--;
            } else if (*value == '{') {
                count++;
            }
        }
        return 0;
    }
    return -1;
}

static ev_tstamp
get_ev_loop_clock()
{
      struct timespec ts;

      if (!clock_gettime (CLOCK_MONOTONIC, &ts))
          return ts.tv_sec + ts.tv_nsec * 1e-9;
      else
          return ev_time();
}
static void
pica_connection_cb(struct ev_loop *loop, ev_io *w, int revents)
{
    struct jrpc_connection *conn;
    struct jrpc_server *server = (struct jrpc_server *) w->data;
    ssize_t bytes_read = 0;
    int ret;
    ev_tstamp now;
    const char *res;

    //get our 'subclassed' event watcher
    conn = (struct jrpc_connection *) w;
    int fd = conn->fd;
    if (conn->pos == (conn->buffer_size - 1)) {
        /* Here we need to check the max size of a Json Obj, if it
         * exceeded the limitation, we will terminate this json request.
         * Or it may cause unlimited memory allocation until exhausting.
         */
        if (conn->buffer_size == MAX_JSON_OBJ_LEN) {
            pica_send_response(conn, "JSON Object:");
            pica_send_response(conn, conn->buffer);
            pica_send_response(conn, "\n");
            sprintf(conn->buffer, "The lenght of Json Object exceeded \
                    the limitation:%d\n",MAX_JSON_OBJ_LEN);
            pica_send_response(conn, conn->buffer);
            return pica_close_connection(loop, w);
        }

        char *new_buffer = (char *)realloc(conn->buffer, conn->buffer_size *= 2);
        if (new_buffer == NULL) {
            perror("Memory error");
            return pica_close_connection(loop, w);
        }
        conn->buffer = new_buffer;
        memset(conn->buffer + conn->pos, 0, conn->buffer_size - conn->pos);
    }
    // can not fill the entire buffer, string must be NULL terminated
    int max_read_size = conn->buffer_size - conn->pos - 1;
    if ((bytes_read = read(fd, conn->buffer + conn->pos, max_read_size)) == -1) {
        perror("read");
        return pica_close_connection(loop, w);
    }
    if (!bytes_read) {
        // client closed the sending half of the connection
        if (server->debug_level) {
            printf("Client closed connection.\n");
        }
        return pica_close_connection(loop, w);
    } else {
        char *end_ptr;
        char separator;
        conn->pos += bytes_read;

_loop: //this loop is used to process all json objs we have read.
        if ((ret = pica_parse_stream(conn->buffer,
            (const char *)(conn->buffer + conn->pos), &end_ptr)) > 0) {
            if (server->debug_level > 1) {
                printf("Found a potential object.\n");
            }

            /* Separate one json obj pica_parse_stream func found. If we don't
             * divide it, jsonrpc_server_execute will process whole buffer, the
             * whole string as a json obj, which may cause json syntax error.
             * */
            separator = *end_ptr;
            *end_ptr = '\0';
            res = jsonrpc_server_execute(jsonrpc_server, (const char *)conn->buffer);
            pica_send_response(conn, res);

            /* sleep 100ms after processing one JSON OBJ if timer is inactive or
             * timeout(timeout_watcher-now) is longer than 0.110s. 0.1 means sleep
             * 0.1s, 0.01 means the time maybe costed until next timer loop.*/
            now = get_ev_loop_clock();
            if (!ev_is_active(&timeout_watcher) ||
                    timeout_watcher.at - now > 0.110) { 
                //This is for test or debug.
                //if (ev_is_active(&timeout_watcher))
                //    printf("now:%lf at:%lf repeat:%lf\n", now, 
                //            timeout_watcher.at,
                //            timeout_watcher.repeat);
                usleep(100000);
            }

            *end_ptr = separator;
            //shift processed request, discarding it
            memmove(conn->buffer, end_ptr, strlen(end_ptr) + 1);
            conn->pos = strlen(end_ptr);
            memset(conn->buffer + conn->pos, 0,
                   conn->buffer_size - conn->pos - 1);
            if (conn->pos != 0)
                goto _loop;
        } else if (ret < 0) {
            // there was an error before the buffer's end
            if (server->debug_level) {
                printf("Invalid JSON received:%s\n",
                       conn->buffer);
            }
            return pica_close_connection(loop, w);
        }
        // just wait for more.
    }
}

static void
pica_accept_cb(struct ev_loop *loop, ev_io *w, int revents)
{
    char s[INET6_ADDRSTRLEN];
    struct jrpc_connection *connection_watcher;

    if (((struct jrpc_server *) w->data)->con_number >= CONNUMLIMITS) {
        /*reject connections when connection number exceding CONNUMLIMITS,
          this connection is in accept queue, will be processed until next 
          ev_loop, and some connections closed.
          */
        ev_io_stop(loop, w);
        printf("ev_io_loop stoped \n");
        return;
    }
    ((struct jrpc_server *) w->data)->con_number++;
    connection_watcher = (jrpc_connection *)malloc(sizeof(struct jrpc_connection));
    struct sockaddr_storage their_addr; // connector's address information
    socklen_t sin_size;
    sin_size = sizeof their_addr;
    connection_watcher->fd = accept(w->fd, (struct sockaddr *) &their_addr, &sin_size);
    if (connection_watcher->fd == -1) {
        perror("accept");
        free(connection_watcher);
    } else {
        if (((struct jrpc_server *) w->data)->debug_level) {
            inet_ntop(their_addr.ss_family,
                      pica_get_in_addr((struct sockaddr *) &their_addr),
                      s, sizeof s);
            printf("server: got connection from %s\n", s);
        }
        ev_io_init(&connection_watcher->io, pica_connection_cb,
                   connection_watcher->fd, EV_READ);
        //copy pointer to struct jrpc_server
        connection_watcher->io.data = w->data;
        connection_watcher->buffer_size = JSONRPCBUFSZ;
        connection_watcher->buffer = (char *)malloc(JSONRPCBUFSZ);
        memset(connection_watcher->buffer, 0, JSONRPCBUFSZ);
        connection_watcher->pos = 0;
        //copy debug_level, struct jrpc_connection has no pointer to struct jrpc_server
        connection_watcher->debug_level = ((struct jrpc_server *) w->data)->debug_level;
        ev_io_start(loop, &connection_watcher->io);
    }
    /*sleep 1s after processing one connection.*/
    //sleep(1);
}

void
pica_jrpc_server_create_thread()
{
    int rc;
    pthread_t thread;
    pthread_attr_t attr;

    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED );
    if(0 != pthread_create(&thread, &attr, ( void * (*)(void *) )pica_jrpc_server_init, NULL )) {
        printf("Can not create pica jsonrpc server thread!\n");
    }
}

int
pica_jrpc_server_init()
{
    int rc;
    jsonrpc_error_t err;

    /* populate RPC methods table */
    jsonrpc_server = jsonrpc_server_open(jsonrpc_plugin_yajl(), NULL);
    err = jsonrpc_server_register_method(jsonrpc_server, JSONRPC_TRUE,
                                         pica_jrpc_get_counters,
                                         "getCounters",
                                         "sql:s");
    err = jsonrpc_server_register_method(jsonrpc_server, JSONRPC_TRUE,
                                         pica_jrpc_prepare_counters,
                                         "prepareCounters",
                                         "when:s,interval:i,repeat:i,write-to-file:b");

    loop = EV_DEFAULT;
    rc = pica_jrpc_server_init_with_ev_loop(&pica_jrpc_server, JSONRPCSVRPRT, loop);
    pica_jrpc_server_run(&pica_jrpc_server);
    return rc;
}

int
pica_jrpc_server_init_with_ev_loop(struct jrpc_server *server, 
                                   int port_number, struct ev_loop *loop)
{
    memset(server, 0, sizeof(struct jrpc_server));
    server->loop = loop;
    server->port_number = port_number;
    char * debug_level_env = getenv("JRPC_DEBUG");
    if (debug_level_env == NULL)
        server->debug_level = 0;
    else {
        server->debug_level = strtol(debug_level_env, NULL, 10);
        printf("JSONRPC-C Debug level %d\n", server->debug_level);
    }
    return pica_jrpc_server_start(server);
}

static int
pica_jrpc_server_start(struct jrpc_server *server)
{
    int sockfd;
    struct addrinfo hints, *servinfo, *p;
    int yes = 1;
    int rv;
    char PORT[6];

    sprintf(PORT, "%d", server->port_number);
    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE; // use my IP

    if ((rv = getaddrinfo(NULL, PORT, &hints, &servinfo)) != 0) {
        fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(rv));
        return 1;
    }

    // loop through all the results and bind to the first we can
    for (p = servinfo; p != NULL; p = p->ai_next) {
        if ((sockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1) {
            perror("server: socket");
            continue;
        }

        if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1) {
            perror("setsockopt");
            exit(1);
        }

        if (bind(sockfd, p->ai_addr, p->ai_addrlen) == -1) {
            close(sockfd);
            perror("server: bind");
            continue;
        }

        break;
    }

    if (p == NULL) {
        fprintf(stderr, "server: failed to bind\n");
        return 2;
    }

    freeaddrinfo(servinfo); // all done with this structure

    if (listen(sockfd, 5) == -1) {
        perror("listen");
        exit(1);
    }
    if (server->debug_level)
        printf("server: waiting for connections...\n");

    ev_io_init(&server->listen_watcher, pica_accept_cb, sockfd, EV_READ);
    server->listen_watcher.data = server;
    ev_io_start(server->loop, &server->listen_watcher);
    return 0;
}

// Make the code work with both the old (ev_loop/ev_unloop)
// and new (ev_run/ev_break) versions of libev.
#ifdef EVUNLOOP_ALL
  #define EV_RUN ev_loop
  #define EV_BREAK ev_unloop
  #define EVBREAK_ALL EVUNLOOP_ALL
#else
  #define EV_RUN ev_run
  #define EV_BREAK ev_break
#endif

void
pica_jrpc_server_run(struct jrpc_server *server)
{
    EV_RUN(server->loop, 0);
}

int
pica_jrpc_server_stop(struct jrpc_server *server)
{
    EV_BREAK(server->loop, EVBREAK_ALL);
    return 0;
}

